package com.atguigu.rabitmq.consumer.listener04;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Date;

/**
 * 消息者限流
 *
 * @author Taeyang
 * @date 2025/4/8
 */
@Component
public class ConsumerDeadLetter {
    // 取消注释，就是测试死信队列的
    // 屏蔽监听器，保证延迟队列中的数据不被消费 @RabbitListener(queues = "direct_queue_01")
    public void consumer01(Message message, Channel channel) {
        byte[] body = message.getBody();
        String msg = new String(body);
        System.out.println("direct_queue_01 ...msg ----> " + msg + " " + new Date());

        // deliveryTag: 是消息的唯一标识符，在消费者处理消息时，用来确认具体哪条消息被处理。
        long deliveryTag = message.getMessageProperties().getDeliveryTag();

        try {
            // 处理后续业务
            int a = 10 / 0;
            System.out.println("处理后续业务");

            channel.basicAck(deliveryTag, true);
        } catch (Exception e) { // 注意：此处需要是Exception，能够捕获到除零异常
            e.printStackTrace();

            try {
                // 直接进入死信队列
                channel.basicNack(deliveryTag, true, false);
            } catch (IOException ex) {
                throw new RuntimeException(ex);
            }

            throw new RuntimeException(e);
        }
    }

    /**
     * 监听死信队列
     *
     * @param message
     * @param channel
     * @throws IOException
     */
    @RabbitListener(queues = "dlx_queue")
    public void consumerDlx(Message message, Channel channel) throws IOException {

        byte[] body = message.getBody();
        String msg = new String(body);

        // 消息标签，标识消息的唯一性
        long deliveryTag = message.getMessageProperties().getDeliveryTag();

        try {
            // 处理后续业务
            System.out.println("dlx_queue 处理后续业务 ----> " + msg + new Date());
            System.out.println("/////////////////// id /////////////////// " + message.getMessageProperties().getHeader("spring_returned_message_correlation"));

            // 模拟失败
            int a = 10 / 0;

            // 正常消费手动应答ack
            channel.basicAck(deliveryTag, true);

        } catch (Exception e) {
            e.printStackTrace();
            // 未成功消费
            try {
                // 第三个参数选择 false，死信队列中的如果还是无法消费便直接清除或者进入其他的死信队列等待人工处理
                // 第三个参数选择 true，如何下面抛出了异常就会持续在该死信队列安装自定义的重试策略进行循环重试，如果没有抛出异常就会反复投递，疯狂循环
                channel.basicNack(deliveryTag, true, false);
            } catch (IOException ex) {
                throw new RuntimeException(ex);
            }
            // 抛出异常就会进入重试，因为相当于变相的没有捕获到异常，触发重试机制
            throw new RuntimeException(e);
        }
    }
}
